package pubsub;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.InputStream;
import java.net.URI;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import topics.Topic;
import topics.TopicSpace;
import topics.TopicSpaceSet;
import topics.expression.TopicExpression;
import topics.expression.TopicExpressionException;
import topics.expression.impl.TopicExpressionImpl;
import topics.impl.SubscriptionTopicListener;
import topics.impl.TopicSpaceSetImpl;
import topics.util.TopicSpaceParser;
import topics.v2004_06.TopicsConstants;
import util.xml.NamespaceContextImpl;

public class WSNotifier {

	private static WSNotifier instance;

	private TopicSpaceSet topicspaceset = null;
	private TopicSpace topicspace = null;
	private static final String TOPICS_FILE = "topics.xml";
	public static final String TOPIC_EXPR_DIALECT_CONCRETE = TopicsConstants.TOPIC_EXPR_DIALECT_CONCRETE;
	public static final String TOPIC_EXPR_DIALECT_FULL = TopicsConstants.TOPIC_EXPR_DIALECT_FULL;
	public static final String TOPIC_EXPR_DIALECT_SIMPLE = TopicsConstants.TOPIC_EXPR_DIALECT_SIMPLE;

	private Log log = LogFactory.getLog(WSNotifier.class);

	public WSNotifier() throws Exception {
		startUp();
	}

	private void startUp() throws Exception {
		File file = new File(TOPICS_FILE);
		InputStream inStream = null;

		try {
			inStream = new FileInputStream(file);
		} catch (FileNotFoundException e) {
			log.fatal("The file "+TOPICS_FILE+" could not be found!", e);
			throw new Exception("The file "+TOPICS_FILE+" could not be found!");
		}

		topicspace = TopicSpaceParser.parse(inStream);

		topicspaceset = new TopicSpaceSetImpl(true);
		topicspaceset.addTopicSpace(topicspace);
	}

	public static synchronized WSNotifier getInstance() throws Exception {
		if (instance != null) {
			return instance;
		} else {
			return new WSNotifier();
		}
	}

	public synchronized void subscribe(Object topic, String dialect, Subscriber sub) throws Exception {
		if(topic == null){
			throw new Exception("the topic/topics cannot be null");
		}
		if(dialect == null){
			throw new Exception("the dialect cannot be null");
		}
		if(sub == null){
			throw new Exception("the subscriber cannot be null");
		}
		Subscription subscription = new Subscription(sub);
		
		for(Topic t:getTopics(topic, dialect)){
			t.addTopicListener(new SubscriptionTopicListener(subscription));
		}
	}

	public synchronized void publish(Object topic, String dialect, Object rawMsg) throws Exception {
		if(topic == null){
			throw new Exception("the topic/topics cannot be null");
		}
		if(dialect == null){
			throw new Exception("the dialect cannot be null");
		}
		if(rawMsg == null){
			throw new Exception("the message cannot be null");
		}
		for(Topic t:getTopics(topic, dialect)){
			t.publish(rawMsg);
		}
	}

	private Topic[] getTopics(Object topic, String topicExpressionDialect) throws Exception {
		if (!(topicExpressionDialect.equals(TopicsConstants.TOPIC_EXPR_DIALECT_SIMPLE) || 
				topicExpressionDialect.equals(TopicsConstants.TOPIC_EXPR_DIALECT_FULL) || 
				topicExpressionDialect.equals(TopicsConstants.TOPIC_EXPR_DIALECT_CONCRETE))) {
			throw new Exception("Unknown dialect!");
		}
		  
		TopicExpression expression = new TopicExpressionImpl((Object) topic, (Object) new NamespaceContextImpl(), new URI(topicExpressionDialect));

		try {
			return topicspaceset.evaluateTopicExpression(expression);
		} catch (TopicExpressionException e) {
			throw new Exception(e.getMessage());
		}
	}

}
